package com.xdhl.fu.fim.consumer;

import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.xdhl.fu.common.core.constant.event.FlightStockSyncConstants;
import com.xdhl.fu.common.redis.service.RedisService;
import com.xdhl.fu.fim.repository.MGFlightScheduleRepository;
import com.xdhl.fu.fim.service.IFlightStockService;
import com.xdhl.fu.flight.info.api.domain.FlightStock;
import com.xdhl.fu.flight.info.api.mongodb.domain.MGFlightSchedule;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import java.util.List;
import java.util.Optional;

/**
 * 航班座位库存同步事件
 *
 * @author 朱力维
 * @time 2022/6/11 20:23
 */
@Component
public class FlightStockSyncEventConsumer {

    private static final Logger log = LoggerFactory.getLogger(FlightStockSyncEventConsumer.class);

    @Value("#{'${spring.application.name}'+'-'+'${spring.profiles.active}'}")
    private String consumerGroup;

    @Value("${fu.rocketmq.namesrv-addr}")
    private String namesrvAddr;

    private DefaultMQPushConsumer consumer;

    @Resource
    private RedisService redisService;

    @Resource
    private IFlightStockService flightStockService;

    @Resource
    private MGFlightScheduleRepository mgFlightScheduleRepository;

    @PostConstruct
    public void init() throws MQClientException {
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);

        //设置consumer所订阅的Topic和Tag，*代表全部的Tag
        consumer.subscribe(FlightStockSyncConstants.TOPIC, "*");

        /**
         * CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，跳过历史消息
         * CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

        consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> doing(list));
        consumer.start();
        log.info("航班库存同步事件消费者启动成功。");
    }

    @PreDestroy
    public void destory() {
        consumer.shutdown();
        log.info("航班库存同步事件消费者已销毁。");
    }

    public ConsumeConcurrentlyStatus doing(List<MessageExt> msgs) {
        ConsumeConcurrentlyStatus consumeConcurrentlyStatus;
        try {
            for (MessageExt msg : msgs) {
                JSONObject data = JSONObject.parseObject(new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET));
                String cacheKey = data.getString("cacheKey");
                String scheduleDataId = data.getString("scheduleDataId");
                if (StringUtils.isNotBlank(cacheKey) && redisService.hasKey(cacheKey + "A") && redisService.hasKey(cacheKey + "B") && redisService.hasKey(cacheKey + "C")) {
                    String headCountStr = redisService.getStringValue(cacheKey + "A");
                    String businessCountStr = redisService.getStringValue(cacheKey + "B");
                    String economyCountStr = redisService.getStringValue(cacheKey + "C");
                    Integer headCount = Integer.valueOf(headCountStr);
                    Integer businessCount = Integer.valueOf(businessCountStr);
                    Integer economyCount = Integer.valueOf(economyCountStr);
                    LambdaQueryWrapper<FlightStock> wrapper = new LambdaQueryWrapper<>();
                    wrapper.eq(FlightStock::getFlightScheduleDataId, scheduleDataId);
                    FlightStock flightStock = flightStockService.getOne(wrapper);
                    flightStock.setHeadSeatRemainderCount(headCount);
                    flightStock.setBusinessSeatRemainderCount(businessCount);
                    flightStock.setEconomySeatRemainderCount(economyCount);
                    flightStockService.updateById(flightStock);

                    // 更新文档数据库库存
                    Optional<MGFlightSchedule> mgFlightScheduleOptional = mgFlightScheduleRepository.findById(scheduleDataId);
                    if (mgFlightScheduleOptional.isPresent()) {
                        MGFlightSchedule mgFlightSchedule = mgFlightScheduleOptional.get();
                        mgFlightSchedule.setFlightStock(flightStock);
                        mgFlightScheduleRepository.save(mgFlightSchedule);
                    }
                }
            }
            consumeConcurrentlyStatus = ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            log.error(e.getMessage());
            consumeConcurrentlyStatus = ConsumeConcurrentlyStatus.RECONSUME_LATER;
            // nothing to do.
        }
        return consumeConcurrentlyStatus;
    }

}
